package com.sailing.lianxi.demoservice.onecusmer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import com.sailing.lianxi.common.Constants;

public class ProducerExample {

	public static void main(String[] args) {
		
	}
	
	/**
	 * 生产数据
	 * @throws InterruptedException
	 * @throws ExecutionException
	 */
	private static void produceData() throws InterruptedException, ExecutionException {
		Producer<String, String> producer = createProducer();
		sendMessages(producer);
	}
	
	/**
	 * 创建生产者
	 * @return
	 */
	private static Producer<String, String> createProducer(){
		Properties props = new Properties();
		//kafka服务器地址  
		props.put("bootstrap.servers",Constants.KAFKA_SERVERS);
		props.put("client.id", "DemoProducer");
		/*  acks=0 生产者不会等待服务端的确认
            acks=1 只会等待leader分区的确认
            acks=all或者acks=-1 等待leader分区和follower分区的确认
         */
		props.put("acks", "all");
		//生产者发送失败后，重试的次数
		props.put("retries", 0);
		//生产者批量提交消息大小--批量提交必须
		props.put("batch.size", 16384); //16M
		 //默认立即发送，这里这是发送的延时毫秒数--批量提交必须
		props.put("linger.ms", 10);
		//生产者缓冲大小，当缓冲区耗尽后，额外的发送调用将被阻塞。时间超过max.block.ms将抛出TimeoutException
        props.put("buffer.memory", 33554432);//32M
        //kafka消息key序列化类 若传入key的值，则根据该key的值进行hash散列计算出在哪个partition上  
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        //kafka消息序列化类 即将传入对象序列化为字节数组  
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        
        //创建kafka的生产者类
       Producer<String, String> producer = new KafkaProducer<String,String>(props);
        return producer;
	}
	
	/**
	 * 发送数据
	 * @param producer
	 * @throws InterruptedException
	 * @throws ExecutionException
	 */
	private static void sendMessages(Producer<String, String> producer) throws InterruptedException, ExecutionException {
		  //发送消息
        for(int i=0;i<1;i++){
        	//封装消息体,这里平均写入４个分区
        	//如果不是创建topic时指定分区数，在生产时创建topic 则按照配置文件中配置的分区数和备份数创建topic
        	int partition =0;//i%4; 
        	String msg = "a";
        	ProducerRecord message = new ProducerRecord<String, String>(Constants.TOPIC_NAME,partition,Integer.toString(i),msg+i); 
        	//发送消息
        	RecordMetadata recordMetadata =(RecordMetadata) producer.send(message,new Callback(){
				public void onCompletion(RecordMetadata metadata,
						Exception exception) {
						if(null!=exception){
							System.out.println("记录的offset在"+metadata.offset());
							System.out.println(exception.getMessage());
						}
					
				}}).get();
        	System.out.println("分区："+recordMetadata.partition()+"offset:"+recordMetadata.offset());
        }
        System.out.println("生产数据完成。。");
	}
}
